package top.jolyoulu.streaming.example

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import top.jolyoulu.utils.JDBCUtils

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description 判断用户的瞬间广告点击率是否超过阈值，超过认为机器人加入黑名单
 */
object Spark01_SparkStreaming_BlackList {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->
        "192.168.88.100:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "Req1",
      "key.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer"
    )

    //获取Kafka数据
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("mockData"), kafkaPara)
    )
    //解析数据
    val adClickData: DStream[AdClickData] = kafkaDataDS.map(
      kafkaData => {
        val data: String = kafkaData.value()
        val datas: Array[String] = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    )
    //使用transform操作rdd
    val ds: DStream[((String, String, String), Int)] = adClickData.transform(
      rdd => {
        //通过JDBC周期性获取黑名单数据
        val blackList = ListBuffer[String]()
        val conn: Connection = JDBCUtils.getConnection
        val pstat: PreparedStatement = conn.prepareStatement("select userid from black_list")
        val rs: ResultSet = pstat.executeQuery()
        while (rs.next()) {
          blackList.append(rs.getString(1))
        }
        rs.close()
        pstat.close()
        conn.close()
        //过滤
        val filterRDD: RDD[AdClickData] = rdd.filter(
          data => {
            //判断点击用户是否在黑名单中
            !blackList.contains(data.user)
          }
        )
        //如果用户不在黑名单中，需要进行数量统计（每个周期的）
        filterRDD.map(
          data => {
            val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
            val day = sdf.format(new Date(data.ts.toLong))
            val user = data.user
            val ad = data.ad
            ((day, user, ad), 1)
          }
        ).reduceByKey(_ + _)
      }
    )

    ds.foreachRDD(
      rdd => {
        rdd.foreach {
          case ((day, user, ad), count) => {
            println(s"${day} ${user} ${ad} ${count}")
            val conn: Connection = JDBCUtils.getConnection
            if (count >= 30) {
              //如果统计数量超过阈值(30)，将用户拉入黑名单
              JDBCUtils.executeUpdate(
                conn,
                """
                  |insert into black_list (userid) values (?)
                  |ON DUPLICATE KEY
                  |UPDATE userid = ?
                  |""".stripMargin,
                Array(user, user)
              )
            } else {
              //如果没超过阈值，将当天的广告点击数量进行跟新
              val rs: Boolean = JDBCUtils.isExist(
                conn,
                """
                  |select *
                  |from user_ad_count
                  |where dt = ? and userid = ? and adid = ?
                  |""".stripMargin,
                Array(day, user, ad)
              )
              //查询统计表数据，如果存在就更新，不存在就新增
              if (rs) {
                JDBCUtils.executeUpdate(
                  conn,
                  """
                    |update user_ad_count set count = count + ?
                    |where dt = ? and userid = ? and adid = ?
                    |""".stripMargin,
                  Array(count, day, user, ad)
                )
                //更新后的点击数据是否超过阈值，如果超过，将用户拉入到黑名单
                val rs2: Boolean = JDBCUtils.isExist(
                  conn,
                  """
                    |select *
                    |from user_ad_count
                    |where dt = ? and userid = ? and adid = ? and count >= 30
                    |""".stripMargin,
                  Array(day, user, ad)
                )
                if (rs2) {
                  JDBCUtils.executeUpdate(
                    conn,
                    """
                      |insert into black_list (userid) values (?)
                      |ON DUPLICATE KEY
                      |UPDATE userid = ?
                      |""".stripMargin,
                    Array(user, user)
                  )
                }
              } else {
                JDBCUtils.executeUpdate(
                  conn,
                  """
                    |insert into user_ad_count (dt,userid,adid,count) values (?,?,?,?)
                    |""".stripMargin,
                  Array(day, user, ad, count)
                )
              }
              conn.close()
            }
          }
        }
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }

  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)
}
